package com.wyg.ep_mail.config;

import com.wyg.ep_mail.common.uti.EmailUtil;
import com.wyg.epcommonbase.common.util.StringUtil;
import com.wyg.epkafkastarter.config.KafkaConfigProperties;
import com.wyg.epkafkastarter.consumer.impl.BatchMailerConsumer;
import com.wyg.epkafkastarter.encoder.MailEncoder;
import com.wyg.epkafkastarter.msg.MailMsg;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.stereotype.Component;

import java.util.List;

/**
 * @Author: WangYongGan
 * @Date: 2022/5/18 17:56
 * @Description:
 */
@Component
@Slf4j
public class KafkaListenerStarter implements InitializingBean {
    public static final String subject = "金融舆情提醒";

    public KafkaListenerStarter(KafkaConfigProperties kafkaConfigProperties) {

    }

    @Override
    public void afterPropertiesSet() throws Exception {
        BatchMailerConsumer.start(records -> {
            records.forEach(record -> {
                final MailMsg mailMsg = MailEncoder.decode(record.value());
                if (mailMsg == null) {
                    log.error("消息格式不正确::" + record.value());
                } else {
                    log.info("消费数据::" + record.value());
                    final List<String> entityList = mailMsg.getEntities();
                    final String to = mailMsg.getTo();
                    String msg = StringUtil.trimList(entityList);
                    msg += "公司有新的消息";
                    EmailUtil.sendEmail(to, subject, msg);
                }

            });
        });
    }
}
